package com.example.kafkastudy.producer;

import com.example.kafkastudy.interceptor.ProducerInterceptorPrefix;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.protocol.types.Field;

import java.util.Properties;

/**
 * @Author: zhudawang
 * @Date: 2021/05/03/7:37 下午
 * @Description:
 */
public class KafkaProducerAnalysis {
    public static final String brokerList = "123.56.141.6:9092";
    public static final String topic = "quickstart-events";

    public static Properties initConfig() {
        Properties props = new Properties();
        props.put("bootstrap.servers", brokerList);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("client.id", "producer.client.id.demo");
        props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, ProducerInterceptorPrefix.class.getName());
        return props;
    }

    public static void main(String[] args) {
        Properties props = initConfig();
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
        for (int i = 0; i < 10; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<>(topic, "hello,Kafka!" + i);
            producer.send(record);
        }
        producer.close();
//        try {
//            //producer.send(record);
//            producer.send(record, new Callback() {
//                @Override
//                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
//                    if (e != null) {
//                        e.printStackTrace();
//                    } else {
//                        System.out.println(record.topic() + "-" + recordMetadata.partition() + ":" + recordMetadata.offset());
//                    }
//                }
//            });
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
    }
}
